package com.iscale.rxjava;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by andychen83 on 2017/3/7.
 */
public class RxJava {
    @Test
    public void ObserverTest() {

        //创建被观察者Observable   上游
        /**除了create()，just()和from()同样可以创建Observable
         *
         * ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，
         * 通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、
         * complete事件和error事件
         */
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });

        Observable<Integer> observable2 =   Observable.just(1,2,3);

        Observable<Integer> observable3 =Observable.fromArray(new Integer[]{1,2,3});


        //第一步：创建观察者Observer
        /**
         *
         * 它理解成两根管道之间的一个机关, 当调用它的dispose()方法时,
         * 它就会将两根管道切断, 从而导致下游收不到事件.
         */
        Observer<Object> observer = new Observer<Object>() {

            @Override
            public void onError(Throwable e) {
                System.out.println("onError");
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete");
            }

            @Override
            public void onSubscribe(Disposable disposable) {
                //调用了subscribe() 方法之后才开始发送事件
                System.out.println("onSubscribe");
            }

            @Override
            public void onNext(Object s) {
                System.out.println(s.toString());
            }
        };



        //建立连接
        observable.subscribe(observer);

        observable2.subscribe(observer);

        observable3.subscribe(observer);
    }

    /**
     * Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
     *
     * Schedulers.newThread() 代表一个常规的新线程
     */
    @Test
    public void ConsumerTest(){
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                System.out.println("Observable thread is : " + Thread.currentThread().getName());
                System.out.println("emit 1");
                emitter.onNext(1);
            }
        });

        Consumer<Integer> consumer = new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println("Observer thread is :" + Thread.currentThread().getName());
                System.out.println("onNext: " + integer);
            }
        };

        observable.subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.io())
                .subscribe(consumer);
    }

    /**
     * map是RxJava中最简单的一个变换操作符了, 它的作用就是对上游发送的每一个事件应用一个函数,
     * 使得每一个事件都按照指定的函数去变化
     */
    @Test
    public void mapTest(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This is result " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println( s);
            }
        });
    }

    /**
     * 上游每发送一个事件, flatMap都将创建一个新的水管, 然后发送转换之后的新的事件,
     * 下游接收到的就是这些新的水管发送的数据. 这里需要注意的是, flatMap并不保证事件的顺序,
     * 也就是图中所看到的, 并不是事件1就在事件2的前面. 如果需要保证顺序则需要使用concatMap.
     */
    @Test
    public void flatMapTest(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                System.out.println(s);
            }
        });
    }
}
